package com.hu.flink12.sensor

import com.hu.entity.SensorReading
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time


/**
 * @Author: hujianjun
 * @Create Date: 2021/1/26 16:01
 * @Describe: 从csv文件读取数据，模拟滚动窗口函数
 */

object TimeWindowSensor {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment


    val inputSocketStream = env.socketTextStream("localhost", 7777)

    val dataStream = inputSocketStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    }).keyBy(new KeySelector[SensorReading, String] {
      override def getKey(in: SensorReading): String = {
        in.id
      }
    }).window(TumblingEventTimeWindows.of(Time.seconds(10))) //这里导入Time包需要是window下的
      .reduce((r1, r2) => SensorReading(r1.id, r1.timeStamp, r1.temperature.min(r2.temperature)))
      .print()
    env.execute("读取传感器csv数据")
  }
}


